package com.mxnavi5.example.DataStream;

import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisClusterConfig;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.http.HttpHost;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

public class MySinkTest {
    public static void main(String[] args) throws Exception {
        //    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        SingleOutputStreamOperator dataStreamSource = env.addSource(new UdfSourceTest()).returns(Integer.class)
                .map(x->x.toString()).returns(String.class);
        dataStreamSource.print();

        //kafka
        Properties sinkProps = new Properties();
        sinkProps.put("bootstrap.servers", "127.0.0.1:9093");
        dataStreamSource.addSink(new FlinkKafkaProducer("topic-name", new SimpleStringSchema(), new Properties()));
        dataStreamSource.addSink(new FlinkKafkaProducer<String>("default-topic-name",
                new KafkaSerializationSchema<String>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(String s, @Nullable Long aLong) {
                        return new ProducerRecord("topic-name", "key" .getBytes(StandardCharsets.UTF_8),
                                s.toString().getBytes(StandardCharsets.UTF_8));
                    }
                },
                sinkProps,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE));


        //redis
        FlinkJedisPoolConfig redisConf = new FlinkJedisPoolConfig.Builder().setHost("localhost").setPort(6379).build();
        dataStreamSource.addSink(new RedisSink<String>(redisConf, new RedisMapper<String>() {
            @Override
            public RedisCommandDescription getCommandDescription() {
                return new RedisCommandDescription(RedisCommand.HSET, "testKey");
            }

            @Override
            public String getKeyFromData(String s) {
                return "testField";
            }

            @Override
            public String getValueFromData(String s) {
                return s;
            }
        }));


        //es
        List<HttpHost> HttpHosts = new ArrayList<>();
        HttpHosts.add(new HttpHost("localhost",9200));
        ElasticsearchSink.Builder<String> esSink = new ElasticsearchSink.Builder<>(HttpHosts, new ElasticsearchSinkFunction<String>() {
            @Override
            public void process(String s, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                Map<String, String> map = new HashMap<>();
                map.put("data", s);
                IndexRequest indexRequest = Requests.indexRequest().index("testIndex").type("testType").source(map);
                requestIndexer.add(indexRequest);
            }
        });
        dataStreamSource.addSink(esSink.build());


        //jdbc
        String sql = "INSERT INTO temperatures (sensor,temp) VALUES (?, ?)";
        SinkFunction<String> jdbcSink = JdbcSink.sink(sql,
                new JdbcStatementBuilder<String>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, String s) throws SQLException {
                        preparedStatement.setString(1, "test");
                        preparedStatement.setString(2, s);
                    }
                },
                new JdbcExecutionOptions.Builder().withBatchIntervalMs(1000).withBatchSize(10).build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                        .withUrl("").withUsername("").withPassword("").build()
        );
        dataStreamSource.addSink(jdbcSink);


        env.execute();


    }

}
